package dm.ass1;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import dm.ass1.io.ProfileWritable;

public class SplitOutputReducer extends Reducer<Text, ProfileWritable, Text, Text> {

	private MultipleOutputs<Text, Text> mos;
	
	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		super.cleanup(context);
		mos.close();
	}


	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		mos = new MultipleOutputs<Text, Text>(context);
	}


	@Override
	protected void reduce(Text key, Iterable<ProfileWritable> value, Context context) throws IOException, InterruptedException {
		Text keyOut   = new Text();
		Text valueOut = new Text(); 
		
		for (ProfileWritable ratingWritable : value) {
			keyOut.set(Integer.toString((ratingWritable.getRater())));
			valueOut.set(ratingWritable.toString());
			context.write(keyOut, valueOut);
		}
	}
	

}
